use std::sync::Arc;
use std::time::Duration;

use async_std::net::TcpStream;
use futures::channel::oneshot;
use futures::SinkExt;

use crate::actors::router::Router;
use crate::message::{
    MessageId, RpcResponse, RpcResponseError, RpcResult, UserMessageApp, UserMessageForward,
};
use crate::prelude::*;
use crate::rpc;
use crate::Result;

#[async_trait::async_trait]
pub trait ListenDelegate: Send + Sync {
    async fn on_started(&self, stream: &mut TcpStreamActor) -> Result<()>;

    async fn on_stoped(&self, stream: &mut TcpStreamActor) -> Result<()>;
}

pub struct TcpStreamActor {
    remote_node_type: String,
    remote_node_name: String,
    addr: Option<Addr<Self>>,
    listen_delegate: Option<Arc<dyn ListenDelegate>>,
    router: Option<Addr<Router>>,
    message_register: Arc<MessageRegister>,
    framed_write: FramedWrite<Compat<TcpStream>, HoneybeeMsgCodec>,
    framed_read: Option<FramedRead<Compat<TcpStream>, HoneybeeMsgCodec>>,
    session: u32,
    rpc_router: rpc::RpcResponseRouter,
}

impl TcpStreamActor {
    pub fn new(
        remote_node_type: &str,
        remote_node_name: &str,
        framed_write: FramedWrite<Compat<TcpStream>, HoneybeeMsgCodec>,
        framed_read: FramedRead<Compat<TcpStream>, HoneybeeMsgCodec>,
        message_register: Arc<MessageRegister>,
        listen_delegate: Option<Arc<dyn ListenDelegate>>,
    ) -> Self {
        Self {
            remote_node_type: remote_node_type.into(),
            remote_node_name: remote_node_name.into(),
            addr: None,
            listen_delegate,
            router: None,
            message_register,
            framed_write,
            framed_read: Some(framed_read),
            session: 1,
            rpc_router: rpc::RpcResponseRouter::new(),
        }
    }

    pub fn remote_node_type(&self) -> &str {
        &self.remote_node_type
    }

    pub fn remote_node_name(&self) -> &str {
        &self.remote_node_name
    }

    pub fn self_addr(&self) -> Addr<Self> {
        self.addr.clone().unwrap()
    }
}

impl std::fmt::Display for TcpStreamActor {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{}@{}", self.remote_node_name, self.remote_node_type)
    }
}

#[async_trait::async_trait]
impl Actor for TcpStreamActor {
    fn name(&self) -> &str {
        "TcpStreamActor"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        self.addr = Some(ctx.address());
        self.router = Some(Router::from_registry().await?);
        info!("{} tcp stream started", self);
        let framed_read = self.framed_read.take().unwrap();
        if let Some(listen_delegate) = self.listen_delegate.clone() {
            let _ = listen_delegate.on_started(self).await?;
        }
        ctx.add_stream(framed_read);
        Ok(())
    }

    async fn stopped(&mut self, _ctx: &mut Context<Self>) {
        if let Some(listen_delegate) = self.listen_delegate.clone() {
            if let Err(err) = listen_delegate.on_stoped(self).await {
                error!("err:{}", err);
            }
        }
        error!("{} tcp stream actor stoped, disconnect", self,);
    }
}

type MsgResult = std::result::Result<HoneybeeRawMsg, anyhow::Error>;
#[async_trait::async_trait]
impl StreamHandler<MsgResult> for TcpStreamActor {
    async fn handle(&mut self, ctx: &mut Context<Self>, msg: MsgResult) {
        let mut rpc_error: Option<(u32, String)> = None;
        match msg {
            Ok(msg) => {
                let session = msg.session;
                if msg.id == RpcResponse::<()>::ID {
                    self.rpc_router.response(msg.session, msg.bytes);
                    return;
                }

                match self.router {
                    Some(ref router) => {
                        let msg_id = msg.id;
                        if let Err(err) = self.message_register.process(ctx.address(), router, msg)
                        {
                            if session != 0 {
                                rpc_error = Some((session, err.to_string()));
                            }
                            error!("{} process message:{} error:{}", self, msg_id, err);
                        }
                    }
                    None => {
                        if session != 0 {
                            rpc_error = Some((session, "remote router is None".into()));
                        }
                        error!("{} process message:{} router is None", self, msg.id);
                    }
                }
            }
            Err(err) => {
                error!("{} receive message error:{}", self, err);
            }
        }
        if let Some((session, err)) = rpc_error {
            let addr = ctx.address();
            let _ = addr.send(SendMsg {
                session,
                msg: RpcResponse::<()>(Err(RpcResponseError::RpcError(err))),
            });
        }
    }

    async fn finished(&mut self, ctx: &mut Context<Self>) {
        error!("{} tcp disconnected", self);
        let _ = ctx.stop(None);
    }
}

pub struct SendMsg<T>
where
    T: HoneybeeMessage,
{
    pub session: u32,
    pub msg: T,
}

impl<T> Message for SendMsg<T>
where
    T: HoneybeeMessage,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<SendMsg<T>> for TcpStreamActor
where
    T: HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendMsg<T>) -> Result<()> {
        if let Err(err) = self
            .framed_write
            .send(HoneybeeMsg {
                session: msg.session,
                msg: &msg.msg,
            })
            .await
        {
            error!(
                "{} session:{} msg:{:?} send_to_peer failed:{}",
                self, msg.session, msg.msg, err
            );
        }

        Ok(())
    }
}

pub struct CallMsg<T: Message<Result = R> + HoneybeeMessage, R: RpcResult>(
    pub oneshot::Sender<std::result::Result<R, RpcResponseError>>,
    pub T,
);

impl<T, R> Message for CallMsg<T, R>
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T, R> Handler<CallMsg<T, R>> for TcpStreamActor
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: CallMsg<T, R>) -> Result<()> {
        let tx = msg.0;
        let msg = msg.1;

        let session = self.session;
        if let Err(err) = self
            .framed_write
            .send(HoneybeeMsg { session, msg: &msg })
            .await
        {
            error!("{} msg:{:?} send_to_peer failed:{}", self, msg, err);
        }
        self.session = self.session.wrapping_add(1);
        if self.session == 0 {
            self.session = self.session.wrapping_add(1);
        }
        self.rpc_router.register::<T, R>(session, tx);

        Ok(())
    }
}

pub struct AsyncCallMsg<
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
>(pub Addr<A>, pub T);

impl<A, T, R> Message for AsyncCallMsg<A, T, R>
where
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<A, T, R> Handler<AsyncCallMsg<A, T, R>> for TcpStreamActor
where
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: AsyncCallMsg<A, T, R>) -> Result<()> {
        let source = msg.0;
        let msg = msg.1;

        let session = self.session;
        if let Err(err) = self
            .framed_write
            .send(HoneybeeMsg { session, msg: &msg })
            .await
        {
            error!("{} msg:{:?} send_to_peer failed:{}", self, msg, err);
        }
        self.session = self.session.wrapping_add(1);
        if self.session == 0 {
            self.session = self.session.wrapping_add(1);
        }
        self.rpc_router.register_async::<A, T, R>(session, source);

        Ok(())
    }
}

#[async_trait::async_trait]
impl RemoteAddrExt for Addr<TcpStreamActor> {
    fn send_remote<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(SendMsg { session: 0, msg })
    }

    async fn call_remote<T, R>(&self, msg: T) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsg(tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        rx.await
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    async fn call_remote_timeout<T, R>(
        &self,
        msg: T,
        dur: Duration,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsg(tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        timeout(dur, rx)
            .await
            .map_err(|_| RpcResponseError::RpcTimeOut(dur.as_millis()))?
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    fn async_call_remote<A, T, R>(&self, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        self.send(AsyncCallMsg(source, msg))
    }
}

impl UserSender for Addr<TcpStreamActor> {
    fn send_to_user<T>(&self, uid: i64, msg: T)
    where
        T: 'static + prost::Message + MessageId,
    {
        let bytes = {
            let mut buf = Vec::with_capacity(msg.encoded_len());
            if let Err(err) = msg.encode(&mut buf) {
                error!("uid:{} send msg:{} err:{}", uid, T::ID, err);
                return;
            }
            buf
        };
        if let Err(err) = self.send_remote(UserMessageForward {
            uid,
            msg_id: T::ID,
            bytes,
        }) {
            error!("uid:{} send msg:{} err:{}", uid, T::ID, err);
        }
    }
}

impl AppUserSender for Addr<TcpStreamActor> {
    fn send_to_app_user<T>(&self, uid: i64, msg: T)
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        match bincode::serialize(&msg) {
            Ok(bytes) => {
                if let Err(err) = self.send_remote(UserMessageApp {
                    uid,
                    msg_id: T::ID,
                    bytes,
                }) {
                    error!("send_remote uid:{} msg:{:?} failed:{}", uid, msg, err);
                }
            }
            Err(err) => error!("send_to_app_user msg:{:?} failed:{}", msg, err),
        }
    }
}
